package org.huangrui.spark.java.core.req;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.Serializable;

/**
 * @Author hr
 * @Create 2024-10-18 19:39
 */
public class HotCategoryTop10_1 {
    public static void main(String[] args) {
        final SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("spark");
        final JavaSparkContext jsc = new JavaSparkContext(conf);
        // TODO 将文件作为数据源，对接RDD
        final JavaRDD<String> dataRdd = jsc.textFile("data/user_visit_action.txt");
        // TODO 1. 将多余的数据进行删除（过滤）处理
        JavaRDD<String> filterRdd = dataRdd.filter(line -> {
            String[] ss = line.split("_");
            return "null".equals(ss[5]);
        });
        // TODO 2. 将过滤后的数据进行分组统计
        //         将过滤后的数据进行结构的转变：(品类， 0， 1，  0)
        filterRdd.mapToPair(line -> {
            String[] ss = line.split("_");
            if (!"-1".equals(ss[6]) && !"null".equals(ss[6])) {
                // TODO 点击数据
                return new Tuple2<>(ss[6],new HotCategory(ss[6], 1L, 0L, 0L));
            } else if (!"null".equals(ss[8])) {
                // TODO 下单数据
                // ((a,b,c), 0, 1, 0)
                // =>
                // (a, 0, 1, 0)
                // (b, 0, 1, 0)
                // (c, 0, 1, 0)
                return new Tuple2<>(ss[8],new HotCategory(ss[8], 0L, 1L, 0L));
            }else {
                // TODO 支付数据
                return new Tuple2<>(ss[10],new HotCategory(ss[10], 0L, 0L, 1L));
            }
        }).reduceByKey((v1, v2) -> {
            v1.setClickCount(v1.getClickCount() + v2.getClickCount());
            v1.setOrderCount(v1.getOrderCount() + v2.getOrderCount());
            v1.setPayCount(v1.getPayCount() + v2.getPayCount());
            return v1;
        }).take(10).forEach(System.out::println);

        jsc.close();
    }
}
// TODO 自定义数据对象
//     1. 实现可序列化接口
//     2. 遵循Bean规范
//     3. 提供无参和全参的构造方法
//     4. 重写toString方法
@Data
@NoArgsConstructor
@AllArgsConstructor
class HotCategory implements Serializable {
    private String categoryId;
    private Long clickCount;
    private Long orderCount;
    private Long payCount;
}